package com.hexiang.rocketmq.producer;


import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.exception.RequestTimeoutException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.RequestCallback;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;

@Component
@Slf4j
public class RpcProducer {

    private static final String TOPIC_NAME = "RPC_HELLO_TOPIC";
    private static final String PRODUCER_GROUP = "rpc-hello-topic-producer-group-001";

    private DefaultMQProducer producer;

    @PostConstruct
    public void start() throws MQClientException {
        producer = new DefaultMQProducer(PRODUCER_GROUP, getAclRPCHook());
        producer.start();
    }

    @PreDestroy
    public void shutdown() {
        producer.shutdown();
    }

    RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("rocketmq_u1", "12345678"));
    }

    public void syncSendRpcMsg(String msgBody) {
        Message message = new Message(TOPIC_NAME, "", msgBody.getBytes(StandardCharsets.UTF_8));
        Message responseMessage = null;
        try {
            responseMessage = producer.request(message, 10000);
        } catch (RequestTimeoutException | MQClientException | RemotingException | MQBrokerException e) {
            log.error("syncSendRpcMsgErr", e);
        } catch (InterruptedException e) {
            log.error("syncSendRpcMsgErr", e);
            Thread.currentThread().interrupt();
        }
        if (null != responseMessage && null != responseMessage.getBody()) {
            log.warn("responseMessage:" + new String(responseMessage.getBody()));
        } else {
            log.warn("responseMessageIsNull");
        }
    }


    public void asyncSendRpcMsg(String msgBody) {
        Message message = new Message(TOPIC_NAME, "", msgBody.getBytes(StandardCharsets.UTF_8));
        try {
            producer.request(message, new RequestCallback() {

                @Override
                public void onSuccess(Message message) {
                    log.info("message:{}", new String(message.getBody()));
                }

                @Override
                public void onException(Throwable e) {
                    log.info("err", e);

                }
            }, 10000);
        } catch (MQClientException | RemotingException | MQBrokerException e) {
            log.error("asyncSendRpcMsgErr", e);
        } catch (InterruptedException e) {
            log.error("asyncSendRpcMsgErr", e);
            Thread.currentThread().interrupt();
        }
    }
}